package cn.com.gf.cmp.am.monitor.trade;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.ObjectMapper;

import cn.com.gf.cmp.am.message.DataType;
import cn.com.gf.cmp.am.message.MessageTransform;
import cn.com.gf.cmp.am.message.RequestMessage;
import cn.com.gf.cmp.am.monitor.entrust.EntrustSourceContext;


/**
 * 成交数据源监控
 */
@Component
public class TradeSourceMonitor {
	
	private final Logger logger = LoggerFactory.getLogger(TradeSourceMonitor.class);
	
	@Resource
	private JdbcTemplate jdbcTemplate;
	
	@Resource(name="tradeKafkaTemplate")
	private KafkaTemplate<String, String> template;
	
	@Value("${spring.kafka.amtrade.topics}")
	private String topic;
	
	@Value("${table.trade}")
	private String tradeTable = "trealdeal";

	@Value("${table.trade.kafka}")
	private String tradeKafkaTable="trealdeal_kafka";
	
	@Value("${stop.time}")
	private int stopTime=153000;

	@Value("${persist.batch.size}")
	private int persistBatchSize=5000;

	//自营8002
    private TradeSourceContext contextZY = new TradeSourceContext(DataType.ZYNumber);
	//资管8003
	//private TradeSourceContext contextZG = new TradeSourceContext(DataType.ZGNumber);


	private String SELECT_SQL_ZY = "";
	//private String SELECT_SQL_ZG = "SELECT 'am trade' as message_type , trade.* FROM "+tradeTable+" trade WHERE l_realdeal_serial_no > ?  and (c_stock_type = '1' or c_stock_type = 'a') and branch_no=8003 order by l_realdeal_serial_no asc";

	private String INSERT_SQL="insert into "+tradeKafkaTable+"(l_date,branch_no,serial_no) values(?,?,?)";
	
	private String DELETE_SQL="delete from  "+tradeKafkaTable+" where l_date<"+DateTime.now().minusDays(7).toString("yyyyMMdd");
	
	
	private ObjectMapper mapper = new ObjectMapper();
	
	
	public String getTopic() {
		return topic;
	}


	public void setTopic(String topic) {
		this.topic = topic;
	}
	
	 
	
	public String getTradeTable() {
		return tradeTable;
	}


	public void setTradeTable(String tradeTable) {
		this.tradeTable = tradeTable;
	}


	public String getTradeKafkaTable() {
		return tradeKafkaTable;
	}


	public void setTradeKafkaTable(String tradeKafkaTable) {
		this.tradeKafkaTable = tradeKafkaTable;
	}


	public int getStopTime() {
		return stopTime;
	}


	public void setStopTime(int stopTime) {
		this.stopTime = stopTime;
	}




	@Scheduled(fixedDelay=10*60*1000)
	public void loadAndProduce() {
		try{
            
			//推送自营8002数据
			Long serialNoZY = contextZY.getSerialNo(jdbcTemplate,tradeKafkaTable);
			String branchNo="";
			String ldate=""; 
				
			SELECT_SQL_ZY = "SELECT 'gw_trade' as message_type,'8815' as branch_no, trade.* FROM "+tradeTable+" trade WHERE serialnum > ? order by serialnum asc";
			INSERT_SQL="insert into "+tradeKafkaTable+"(l_date,branch_no,serial_no) values(?,?,?)";
			 
			List<Map<String,Object>> dataList = jdbcTemplate.queryForList(SELECT_SQL_ZY, serialNoZY);
			MessageTransform tran=new MessageTransform();
			List<Object[]> insertList=new ArrayList<Object[]>();
			for(Map<String,Object> data : dataList){
				String stock_account = data.get("regid").toString();
				RequestMessage message =tran.requestTransform(data);
				//template.send(topic, stock_account, mapper.writeValueAsString(message));
				logger.debug("trade:"+mapper.writeValueAsString(message));
				serialNoZY = ((BigDecimal)data.get("serialnum")).longValue();
				branchNo=data.get("branch_no").toString();
				ldate=data.get("knocktime").toString().substring(0,8);
				insertList.add(new Object[]{ldate,branchNo,serialNoZY});
				contextZY.setSerialNo(serialNoZY);

				if(insertList.size()==persistBatchSize){
					jdbcTemplate.batchUpdate(INSERT_SQL, insertList);
					insertList.clear();
				}

			}

			
			
			//资管8003推送数据
			/*Long serialNoZG = contextZG.getSerialNo(jdbcTemplate,tradeKafkaTable);
			SELECT_SQL_ZG = "SELECT 'am trade' as message_type , trade.* FROM "+tradeTable+" trade WHERE l_realdeal_serial_no > ?  and (c_stock_type = '1' or c_stock_type = 'a') and branch_no=8003 order by l_realdeal_serial_no asc";
			  
			dataList = jdbcTemplate.queryForList(SELECT_SQL_ZG, serialNoZG);

			
			for(Map<String,Object> data : dataList){
				String stock_account = data.get("vc_stockholder_id").toString();
				RequestMessage message =tran.requestTransform(data);
				//logger.debug("topic"+topic+" stock_account"+stock_account);
				template.send(topic, stock_account, mapper.writeValueAsString(message));
				logger.debug(mapper.writeValueAsString(message));
				serialNoZG = ((BigDecimal)data.get("l_realdeal_serial_no")).longValue();
				branchNo=data.get("branch_no").toString();
				ldate=data.get("l_date").toString();
				insertList.add(new Object[]{ldate,branchNo,serialNoZG});
				contextZG.setSerialNo(serialNoZG);
			}*/
			
			if(!insertList.isEmpty()){
				 jdbcTemplate.batchUpdate(INSERT_SQL, insertList);
				 insertList.clear();
			}
			
			// 程序关闭前5分钟  删除4天前数据
			DateTime now=DateTime.now();
			DateTime stopDateTime=DateTime.parse(String.format("%06d", stopTime),DateTimeFormat.forPattern("HHmmss")).withDate(now.getYear(), now.getMonthOfYear(), now.getDayOfMonth()).minusMinutes(5);
		
		    if(now.isAfter(stopDateTime)){  
		      DELETE_SQL="delete from  "+tradeKafkaTable+" where branch_no='8815' and l_date<"+DateTime.now().minusDays(4).toString("yyyyMMdd");
			  jdbcTemplate.execute(DELETE_SQL);
		    }
		}catch(Exception e){
			logger.error(e.getMessage(),e);
		}finally{
			
		}
	}
	
	 
}
